#include "StdAfx.h"
#include "GroupAggregationOperator.h"
#include "LineageSynopsis.h"
#include "QueueEntity.h"
#include "DocumentIdentifierGenerator.h"
#include "RelationSynopsis.h"
#include "ElementIterator.h"

GroupAggregationOperator::GroupAggregationOperator(void)
{
	int lineageNumber = 1;
	this->lineageSynopsis.reset(new LineageSynopsis(lineageNumber));
	this->relationSynopsis.reset(new RelationSynopsis());
	this->aggregationAttributeName = "";
	this->groupByAttributeName = "";
}


GroupAggregationOperator::~GroupAggregationOperator(void)
{
}
void GroupAggregationOperator::execution(int executionTimes)
{

#ifdef DEBUG
	std::cout<<"===================operator begin================="<<std::endl;
	std::cout<<"operatorid : "<<this->getId()<<std::endl;
#endif
	assert(this->getInputQueueList().size()==1);
	assert(this->getOutputQueueList().size()==1);
	boost::shared_ptr<QueueEntity>inputQueue = this->getInputQueueList().front();
	boost::shared_ptr<QueueEntity>outputQueue = this->getOutputQueueList().front();
	
	for(int i = 0; i<executionTimes ;i++)
	{
		Element inputElement;
		if(outputQueue->isFull())
		{
			break;
		}
		if(inputQueue->isEmpty())
		{
			break;
		}
		inputQueue->dequeue(inputElement);

		if(inputElement.mark == PLUS_MARK)
		{
			
			Document& inputDocument = inputElement.document;
			assert(inputDocument.hasField(this->groupByAttributeName.c_str()));
			//std::cout<<inputDocument<<std::endl;
			if(this->relationSynopsis->isFull())
			{
				throw runtime_error("left synopsis is full");
			}
			//insert into the relation synopsis
			this->relationSynopsis->insertElement(inputElement);
			
			//deal with the groupmap, if the group already exists, insert the element. 
			//if not exists, create the group and insert the element.
			std::string groupByAttributeValue = inputDocument.getField(this->groupByAttributeName).valuestr();
			std::set<DocumentId>currentGroupSet;
			std::map<std::string, std::set<DocumentId> > ::iterator it = groupMap.find(groupByAttributeValue);
			if(it!=this->groupMap.end())
			{// group already exist
				 std::set<DocumentId>& groupSet =  it->second;
				 currentGroupSet = groupSet;
				 groupSet.insert(inputElement.id);
			}
			else
			{// group not exist
				std::set<DocumentId> groupSet;
				groupSet.insert(inputElement.id);
				this->groupMap.insert(make_pair(groupByAttributeValue,groupSet));
				currentGroupSet = groupSet;
			}
			//calculate the aggregation value
			boost::shared_ptr<SetElementIterator> elementIterator;
			elementIterator.reset(new SetElementIterator(this->relationSynopsis,currentGroupSet));
			elementIterator->initial();
			Element groupElement;
			double average=0,sum=0,count=0;
			while((!outputQueue->isFull())&&elementIterator->getNext(groupElement))
			{
				assert(groupElement.document.getField(this->aggregationAttributeName).isNumber());
				double dl = groupElement.document.getField(this->aggregationAttributeName).Number();
				sum += dl;
				count ++;
			}   
			average = sum/count;
			//generate the result element
			Document newDocument;
			DocumentBuilder newDocumentBuilder;
			newDocumentBuilder.appendElements(inputDocument);
			Element newElement;
			if(this->aggregateOperation==AggregateOperation::AVG)
			{
				newDocumentBuilder.appendNumber(this->resultAttributeName,average);
			}
			else if(this->aggregateOperation==AggregateOperation::SUM)
			{
				newDocumentBuilder.appendNumber(this->resultAttributeName,sum);
			}
			else if(this->aggregateOperation==AggregateOperation::COUNT)
			{
				newDocumentBuilder.appendNumber(this->resultAttributeName,count);
			}	
			newDocument = newDocumentBuilder.obj();
			newElement.id = DocumentIdentifierGenerator::generateNewDocumentIdentifier();
			newElement.document = newDocument;
			newElement.mark = PLUS_MARK;
			newElement.timestamp = inputElement.timestamp;

			//insert into lineage synopsis
			if(this->lineageSynopsis->isFull())
			{
				throw runtime_error("lineage synopsis is full");
			}
			Lineage lineage;
			lineage.lineageNumber = 1;
			lineage.lineageDocumentId[0] = inputElement.id;
			this->lineageSynopsis->insertLineage(lineage,newElement);

			//generate new output to the output queue
			if(outputQueue->isFull())
			{
				throw std::runtime_error("output queue is full");
			}
			outputQueue->enqueue(newElement);

		}
		else if(inputElement.mark  == MINUS_MARK)
		{
			//delete from relation synopsis
			relationSynopsis->deleteElement(inputElement);
			//delete from group map
			std::string groupByAttributeValue = inputElement.document.getField(this->groupByAttributeName).String();
			std::map<std::string, std::set<DocumentId> >::iterator it = this->groupMap.find(groupByAttributeValue);
			assert(it!= this->groupMap.end());
			std::set<DocumentId>& documentIdSet = it->second;
			documentIdSet.erase(inputElement.id);

			//delete from lineage synopsis
			Element elementInSynopsis;
			Lineage lineage;
			lineage.lineageNumber = 1;
			lineage.lineageDocumentId[0] = inputElement.id;
			int ret = lineageSynopsis->getAndDeleteElement(lineage,elementInSynopsis);
			if(ret==false)
			{
				//lineage not found, the minus tuple has already been generated
				//we needn't generate the minus tuple again
				continue;
				//throw std::runtime_error("lineage not exist");
			}
			//generate minus element
			Element outputElement;
			outputElement.mark = MINUS_MARK;
			outputElement.document = elementInSynopsis.document;
			outputElement.timestamp = inputElement.timestamp;
			outputElement.id = elementInSynopsis.id;
			outputQueue->enqueue(outputElement);
		}
	}
#ifdef DEBUG
	std::cout<<"===================operator over================="<<std::endl;
#endif
}
void GroupAggregationOperator::setGroupByAttributeName(std::string groupByAttributeName)
{
	this->groupByAttributeName = groupByAttributeName;
}
void GroupAggregationOperator::setAggregationAttributeName(std::string aggregationAttributeName)
{
	this->aggregationAttributeName = aggregationAttributeName;

}

void GroupAggregationOperator::setResultAttributeName(std::string resultAttributeName)
{
	this->resultAttributeName = resultAttributeName;
}

void GroupAggregationOperator::setAggregateOperation(AggregateOperation& aggregateOperation)
{
	this->aggregateOperation = aggregateOperation;
}
